package com.atcyj.gamll.realtime.common.util;

import com.alibaba.fastjson.JSONObject;
import com.atcyj.gamll.realtime.common.constant.Constant;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

/**
 * @author: cyj
 * @date: 2024/8/15
 */
@Slf4j
public class HBaseUtil {
    /**
     * 获取HBase同步连接
     *
     * @return
     * @throws IOException
     */
    public static Connection getConnection() throws IOException {
        // 1. Configuration 来配置连接Hbase的信息
        Configuration conf = new Configuration();
        conf.set("hbase.zookeeper.quorum", Constant.HBASE_ZOOKEEPER_QUORUM);
        // 端口号是2181 ，可以不用配置
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        // 2. 可以把Hbase的 hbase-site.xml 配置表放到resource中也可以
        return ConnectionFactory.createConnection(conf);
    }

    /**
     * 关闭同步连接
     *
     * @param connection
     */
    public static void closeConnection(Connection connection) {
        if (connection != null && !connection.isClosed()) {
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return;
    }


    /**
     * 创建表
     *
     * @param connection
     * @param hbaseNamespace
     * @param table
     * @param colFamilies
     * @throws IOException
     */
    public static void createTable(Connection connection, String hbaseNamespace, String table, String... colFamilies) throws IOException {

        Admin admin = connection.getAdmin();

        try {
            TableName tableName = TableName.valueOf(hbaseNamespace, table);
            if (admin.tableExists(tableName)) {
                log.info("=================建表检查===============" + hbaseNamespace + ":" + table + " 表已存在");
                return;
            }
            // 表的描述
            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName);
            for (String colFamily : colFamilies) {
                // 列的描述
                ColumnFamilyDescriptor columnFamilyDescriptor = ColumnFamilyDescriptorBuilder.of(Bytes.toBytes(colFamily));
                tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptor);
            }
            admin.createTable(tableDescriptorBuilder.build());
            admin.close();
            log.info("=================建表操作===============" + hbaseNamespace + ":" + table + " 表创建成功");
        } catch (IOException e) {
            log.error("==================建表操作==============" + hbaseNamespace + ":" + table + " 表创建失败");

        }

    }

    /**
     * 删除表
     *
     * @param connection
     * @param namespace
     * @param table
     * @throws IOException
     */
    public static void dtopTable(Connection connection, String namespace, String table) throws IOException {

        Admin admin = connection.getAdmin();
        try {
            TableName tableName = TableName.valueOf(Bytes.toBytes(namespace), Bytes.toBytes(table));
            if (!admin.tableExists(tableName)) {
                log.info("================删表检查================" + namespace + ":" + table + " 表不存在");
                return;
            }
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
            admin.close();
            log.info("===================删表操作=============" + namespace + ":" + table + " 表删除成功");
        } catch (IOException e) {
            log.error("===================删表操作=============" + namespace + ":" + table + " 表删除失败");

        }
    }

    /**
     * 插入一整行数据
     *
     * @param connection
     * @param namespace
     * @param tableName
     * @param rowKey
     * @param colFamilies
     * @param data        写入的数据
     * @throws IOException
     */
    public static void putRow(Connection connection, String namespace, String tableName, String rowKey, String colFamilies, JSONObject data) throws IOException {

        // 1. 获取table
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));

        try {
            // 2. 创建写入对象
            Put put = new Put(Bytes.toBytes(rowKey));
            for (String key : data.keySet()) {
                String value = data.getString(key);
                if (value != null && value.length() != 0) {
                    put.addColumn(Bytes.toBytes(colFamilies), Bytes.toBytes(key), Bytes.toBytes(value));
                }
            }
            // 3、 写入数据
            table.put(put);
        } catch (IOException e) {
            log.error("====================================插入数据失败");
        } finally {
            table.close();
        }

    }

    /**
     * 删除一整行数据
     *
     * @param connection
     * @param namespace
     * @param tableName
     * @param rowKey
     * @throws IOException
     */
    public static void deleteRow(Connection connection, String namespace, String tableName, String rowKey) throws IOException {
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));
        try {

            // 删除一整行
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            //delete.addColumn(Bytes.toBytes(colFamilies), Bytes.toBytes(colName));
            table.delete(delete);
        } catch (IOException e) {
            log.error("==================================删除数据失败");
        } finally {
            table.close();
        }

    }

}
